/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.queue.impl;

import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicLong;

import org.sabayframework.channel.Channel;
import org.sabayframework.log.Log;

/**
 * Centralized execution queue. All worker threads share one activity queue.
 * 
 * @author Alex Antonau
 * */
public final class CentralizedExecutionQueue extends BaseExecutionQueue implements Runnable {

	private final ArrayDataQueue<ActivityContextImpl<?,?>> queue;
	private final Monitor monitor;
	private final int pollTimeMillis;
	private volatile int waitingThreads;
	private volatile boolean hasPollThread;
	
	protected final AtomicLong activityTime;
	protected final AtomicLong activityCalls;
	protected final AtomicLong activityCount;
	
	public CentralizedExecutionQueue() {
		this(Runtime.getRuntime().availableProcessors());
	}
	public CentralizedExecutionQueue(ThreadFactory threadFactory) {
		this(threadFactory, Runtime.getRuntime().availableProcessors());
	}
	
	public CentralizedExecutionQueue(int threadNumber) {
		this(threadNumber, 10000);
	}
	public CentralizedExecutionQueue(int threadNumber, int queueSize) {
		this(new DefaultThreadFactory(), threadNumber, queueSize);
	}
	public CentralizedExecutionQueue(ThreadFactory threadFactory, int threadNumber) {
		this(threadFactory, threadNumber, 10000);
	}
	public CentralizedExecutionQueue(ThreadFactory threadFactory, int threadNumber, int queueSize) {
		this(threadFactory, threadNumber, queueSize, 10);
	}
	
	public CentralizedExecutionQueue(ThreadFactory threadFactory, int threadNumber, int queueSize, int pollTimeMillis) {
		super(threadFactory);
		this.activityTime = new AtomicLong();
		this.activityCalls = new AtomicLong();
		this.activityCount = new AtomicLong();
		this.active = true;
		this.pollTimeMillis = pollTimeMillis;
		
		queue = new ArrayDataQueue<ActivityContextImpl<?,?>>(queueSize);
		this.monitor = new Monitor();
		start(threadNumber);
	}
	
	/**
	 * Returns number of activities currently in the queue
	 * */
	public int size() {
		return queue.size();
	}
			
	@Override
	public boolean isEmpty() {
		return queue.isEmpty();
	}

	@Override
	protected <I, O> void assignToWorker(ActivityContextImpl<I, O> ctx) throws InterruptedException {
		queue.add(ctx);
		monitor.changed();
		Log.debug("Queued ", ctx.getActivityName());
	}

	@Override
	public void channelChanged(Channel<?> channel) {
		monitor.changed();
	}
	private ActivityContextImpl<?, ?> pollQueue(ArrayDataQueue<ActivityContextImpl<?, ?>> queue, int size){
		//loop through the queue and look for ready activities
		while(size-->0){
			//break the loop if the queue is blocked
			final ActivityContextImpl<?,?> a = queue.tryPoll();
			if(a == null) return null;
			if(a.isReady()) {					
				return a;
			}
			if(!a.isDone()) queue.add(a);
			else debriefActivity(a);
		};
		return null;
	}

	private ActivityContextImpl<?,?> nextReady() throws InterruptedException {
		while(active){
			
			//loop through the queue and look for ready activities
			int size = queue.size();
			final ActivityContextImpl<?,?> a = pollQueue(queue, size);
			if(a != null) {
				if(!hasPollThread && waitingThreads>0) monitor.changed();
				return a;
			}
			monitor.lock();
			try{
				waitingThreads++;
				if(!hasPollThread) {
					//only one thread periodically polls the queue
					hasPollThread = true;
					monitor.waitChanged(pollTimeMillis);
					hasPollThread = false;
				}
				else monitor.waitChanged();
				waitingThreads--;
			}
			finally{
				monitor.unlock();
			}
		}
		throw new InterruptedException();
	}
	
	private void debriefActivity(ActivityContextImpl<?,?> a){
		//collect stats
		this.activityCount.incrementAndGet();
		this.activityCalls.addAndGet(a.getCallCount());
		this.activityTime.addAndGet(a.getExecutionTime());
		a.afterRemoving(this);
		Log.debug("Unqueued ", a.getActivityName());
	}
	
	private void returnActivity(ActivityContextImpl<?,?> a){
		if(!a.isDone()) queue.add(a);
		else debriefActivity(a);
	}
		
	public void run() {
		while(active){
			try{
				ActivityContextImpl<?,?> a = nextReady();
				runActivity(a, false);
				returnActivity(a);
			} 
			catch (InterruptedException e) {}
			catch (Throwable t) {
				lastError = t;
			}
		}
	}
	
	
	@Override
	public Runnable createWorker(int workerNo) {
		return this;
	}
			
	public long getActivityTime() {
		return activityTime.get();
	}
	public long getActivityCalls() {
		return activityCalls.get();
	}
	public long getActivityCount() {
		return activityCount.get();
	}
	public void clearStats(){
		activityTime.set(0);
		activityCalls.set(0);
		activityCount.set(0);
	}
	
	@Override
	public String toString() {
		return getStateInfo();
	}

	public String getStateInfo(){
		StringBuilder sb=new StringBuilder();
		sb.append("Active: "+active);
		sb.append("\nNumber of threads: "+workerCount());
		Object[] qac=queue.toArray();
		sb.append("\n\nQueued activities: "+qac.length);
		for(Object o:qac){
			ActivityContextImpl<?,?> a = (ActivityContextImpl<?,?>)o; 
			dumpActivity(sb, a);
		}
		for(Thread t:workerThreads){
			sb.append("\n\nThread: "+t.getName());
			StackTraceElement[] stack = t.getStackTrace();
			for(StackTraceElement e:stack){
				sb.append("\n ").append(e.toString());
			}
		}
		final Throwable error = lastError;
		if(error!=null){
			sb.append("\n\nLast error: ");
			StackTraceElement[] stack = error.getStackTrace();
			for(StackTraceElement e:stack){
				sb.append("\n").append(e.toString());
			}
		}
		return sb.toString();
	}
	private void dumpActivity(StringBuilder sb, ActivityContextImpl<?,?> a){
		if(a!=null){
			sb.append("\nActivity: "+a.getActivity().toString());
			sb.append("\n\tCall count: "+a.getCallCount());
			sb.append("\n\tRun time: "+a.getExecutionTime());
		}
	}

}
